package com.yujiahao.bigdata.sql

import org.apache.commons.lang.mutable.Mutable
import org.apache.spark.api.java.function
import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions}
import org.apache.spark.sql.expressions.Aggregator

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object SparkSQL01_Hive_Request {
  def main(args: Array[String]): Unit = {

    //这个要使用hadoop中的用户名，不然会报权限不够错误：user_visit_action
    System.setProperty("HADOOP_USER_NAME", "yujiahao")
    //创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .enableHiveSupport()
      .master("local[*]")
      .appName("sql")
      .getOrCreate()
    spark.sql(" use atguigu0819")
    //TODO 1、将数据补充完整，关联三张表，连接在一起。
    //TODO 2、将区域和商品作为整体，统计点击数量。 (( 区域, 商品 ), sum)
    //TODO 3、将统计结果进行结构的转换。(( 区域, 商品 ), sum) => (区域, ( 商品, sum))
    //TODO 4、将转换后的数据根据区域进行分组。(区域, Iter[( 商品, sum),( 商品1, sum1),( 商品2, sum2)])
    //TODO 5、将分组后的数据按照点击数进行排序，取前三名

    spark.sql(
      """
        |select
        |*
        |from (select
        |*,
        |rank() over(partition by t2.area order by t2.cliCnt desc) as rank
        |from (select
        |t1.area,
        |t1.product_name,
        |count(*) as cliCnt
        |from (select
        |a.*,
        |p.product_name,
        |c.area,
        |c.city_name
        |from user_visit_action a
        |join product_info p on a.click_product_id = p.product_id
        |join city_info c on a.city_id = c.city_id
        |where a.click_product_id != -1) t1
        |group by t1.area ,t1.product_name) t2)t3
        |where rank <= 3
        |""".stripMargin).show()

    //TODO 2、关闭资源
    spark.stop()
  }


}
